package com.xr.util;

import com.xr.service.IBatchProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.List;

/**
 * 批量处理工具类
 * <p>
 * 用于分批处理任务，如：
 * 1、推送中，单次最大允许推送1000个用户
 * 2、其他的因数据量大，需要分批执行的场景
 * <p>
 * Created by Arnold on 2018/6/12.
 */
public class BatchProcessUtil<T> {

  protected static Logger logger = LoggerFactory.getLogger(BatchProcessUtil.class);

  protected static final String BATCH_LOG_PATTERN = "batch:%s# process finished, size:%s, start index:%s, end index:%s, total task size:%s.";

  /**
   * 同步批量处理
   * <p>
   * 如果处理出错，未处理的将不会再执行。
   *
   * @param taskList  需要分批的任务列表
   * @param batchSize 批次大小
   * @param processor 任务处理器
   * @throws Exception
   */
  public static void sync(final List taskList, final int batchSize, final IBatchProcessor processor) throws Exception {
    if (taskList == null || taskList.size() == 0) {
      return;
    }

    int taskSize = taskList.size();
    final int maxIdx = taskSize - 1;

    int count = 1; // 当前批次
    int startIdx = 0;
    int endIdx = (batchSize >= taskSize ? taskSize : batchSize) - 1;
    while (startIdx <= maxIdx) {
      processor.process(count, taskList.subList(startIdx, endIdx + 1));
      startIdx = startIdx + batchSize;
      if (startIdx > maxIdx) {
        break;
      }
      logger.info(String.format(BATCH_LOG_PATTERN, count, batchSize, startIdx, endIdx, taskSize));
      endIdx = (maxIdx - endIdx > batchSize ? endIdx + batchSize : maxIdx);
      count++;
    }
  }

  /**
   * 异步批量处理
   * <p>
   * 如果某个批次数据处理出错，不影响其他数据的处理。
   *
   * @param taskList  需要分批的任务列表
   * @param batchSize 批次大小
   * @param processor 任务处理器
   * @throws Exception
   */
  public static void async(final List taskList, final int batchSize, final IBatchProcessor processor) throws Exception {
    if (taskList == null || taskList.size() == 0) {
      return;
    }

    int taskSize = taskList.size();
    final int maxIdx = taskSize - 1;

    int count = 1; // 当前批次
    int startIdx = 0;
    int endIdx = (batchSize >= taskSize ? taskSize : batchSize) - 1;
    while (startIdx <= maxIdx) {
      final int finalStartIdx = startIdx;
      final int finalEndIdx = endIdx;
      final int finalCount = count;
      ThreadPoolUtil.submit(() -> {
        List subList = taskList.subList(finalStartIdx, finalEndIdx + 1);
        try {
          processor.process(finalCount, subList);
        } catch (Exception e) {
          logger.error(String.format("execute batch:%s# failed, message:%s", finalCount, e.getMessage()));
        }
      });
      logger.info(String.format(BATCH_LOG_PATTERN, count, batchSize, startIdx, endIdx, taskSize));
      startIdx = startIdx + batchSize;
      if (startIdx > maxIdx) {
        break;
      }
      endIdx = (maxIdx - endIdx > batchSize ? endIdx + batchSize : maxIdx);
      count++;
    }
  }
}
